package cn.tedu.stream.window

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow

/**
 * @author Amos
 * @date 2022/5/23
 */

object StreamCountWindowDemo {
  def main(args: Array[String]): Unit = {
    // 环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 构建socket数据源 hello 4  hello 5
    val source: DataStream[String] = env.socketTextStream("hadoop01", 9999)
    // 数据处理
    import org.apache.flink.api.scala._
    val wordAndCount = source.map(x => {
      val fields = x.split(" ")
      (fields(0), fields(1).toInt)
    })

    val keyedStream: KeyedStream[(String, Int), Tuple] = wordAndCount.keyBy(0)
    // countWindow的滚动窗口，要求每达到5条数据统计一次
//    val windowStream: WindowedStream[(String, Int), Tuple, GlobalWindow] = keyedStream.countWindow(5)

    // 每移动3条记录，计算过去5条记录的数据
    val windowStream: WindowedStream[(String, Int), Tuple, GlobalWindow] = keyedStream.countWindow(5, 3)

    val result = windowStream.sum(1)
    result.print()
    env.execute()


  }

}
